package com.swak.consistency.reliablemsg.storage.impls;

import java.time.LocalDateTime;
import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;

import javax.sql.DataSource;

import org.springframework.beans.factory.InitializingBean;

import com.swak.App;
import com.swak.consistency.core.execption.ConsistencyException;
import com.swak.consistency.reliablemsg.context.ReliableMsgLog;
import com.swak.consistency.reliablemsg.storage.ReliableMsgLogStorage;
import com.swak.db.DbStorage;
import com.swak.utils.CollectionUtils;
import com.swak.utils.IOUtils;
import com.swak.utils.Ints;
import com.swak.utils.StringUtils;

public class ReliableMsgLogStorageForMysql extends DbStorage implements ReliableMsgLogStorage, InitializingBean {

	private String module;
	private String tableName;
	private String failTableName;

	public ReliableMsgLogStorageForMysql(DataSource dataSource, String module) {
		super(dataSource);
		this.module = module;
		this.tableName = "RM_" + this.module + "_LOG";
		this.failTableName = "RM_" + this.module + "_FAIL";
	}

	@Override
	public void afterPropertiesSet() throws Exception {
		String tables = IOUtils.readAsText(this.getClass(), "tables.sql");
		tables = StringUtils.replace(tables, "${tableName}", this.tableName);
		tables = StringUtils.replace(tables, "${failTableName}", this.failTableName);
		String[] sqls = StringUtils.split(tables, ";");
		for (String sql : sqls) {
			this.executeUpdate(sql);
		}
	}

	@Override
	public ReliableMsgLog get(Long msgId) {
		String selectSql = "SELECT * FROM " + tableName + " WHERE MSG_ID = ? ";
		List<Map<String, Object>> list = executeQuery(selectSql, msgId);
		if (list != null && list.size() > 0) {
			return list.stream().filter(Objects::nonNull).map(this::buildResultMap).findFirst().orElse(null);
		}
		return null;
	}

	@Override
	public void create(ReliableMsgLog reliableMsg) {
		String sql = "INSERT INTO " + tableName
				+ "(MSG_ID,PHASE,RETRIED_TIMES,MAX_RTS,VERSION,CREATE_TIME,UPDATE_TIME,MESSAGE,QUEUE,QUEUE_TYPE)"
				+ " VALUES(?,?,?,?,?,?,?,?,?,?)";
		try {
			final byte[] serialize = App.serialize(reliableMsg.getMessage());
			executeUpdate(sql, reliableMsg.getMsgId(), reliableMsg.getPhase(), reliableMsg.getRetryTimes(),
					reliableMsg.getMaxRts(), reliableMsg.getVersion(), reliableMsg.getCreateTime(),
					reliableMsg.getUpdateTime(), serialize, reliableMsg.getQueue(), reliableMsg.getQueueType());
		} catch (Exception e) {
			throw new ConsistencyException(e);
		}
	}

	@Override
	public void updatePhase(ReliableMsgLog reliableMsg) {
		try {
			String sql = "UPDATE " + tableName + " SET PHASE = ?, UPDATE_TIME = ?  WHERE MSG_ID = ?  ";
			executeUpdate(sql, reliableMsg.getPhase(), new Date(), reliableMsg.getMsgId());
		} catch (Exception e) {
			throw new ConsistencyException(e);
		}
	}

	@Override
	public void remove(ReliableMsgLog reliableMsg) {
		try {
			String sql = "DELETE FROM " + tableName + " WHERE MSG_ID = ? AND PHASE != 0 ";
			executeUpdate(sql, reliableMsg.getMsgId());
		} catch (Exception e) {
			throw new ConsistencyException(e);
		}
	}

	private ReliableMsgLog buildResultMap(final Map<String, Object> map) {
		ReliableMsgLog transLog = new ReliableMsgLog();
		transLog.setMsgId((Long) map.get("MSG_ID"));
		transLog.setPhase((Integer) map.get("PHASE"));
		transLog.setRetryTimes((Integer) map.get("RETRIED_TIMES"));
		transLog.setMaxRts((Integer) map.get("MAX_RTS"));
		transLog.setVersion((Integer) map.get("VERSION"));
		transLog.setCreateTime((LocalDateTime) map.get("CREATE_TIME"));
		transLog.setUpdateTime((LocalDateTime) map.get("UPDATE_TIME"));
		transLog.setQueue((String) map.get("QUEUE"));
		transLog.setQueueType((String) map.get("QUEUE_TYPE"));
		byte[] bytes = (byte[]) map.get("MESSAGE");
		try {
			final Object message = App.deserialize(bytes);
			transLog.setMessage(message);
		} catch (Exception e) {
			throw new ConsistencyException(e);
		}
		return transLog;
	}

	@Override
	public List<ReliableMsgLog> queryDelayMessage(Date delay) {
		String selectSql = "SELECT * FROM " + tableName + " WHERE UPDATE_TIME < ? LIMIT 20";
		List<Map<String, Object>> list = executeQuery(selectSql, delay);
		if (CollectionUtils.isNotEmpty(list)) {
			return list.stream().filter(Objects::nonNull).map(this::buildResultMap).collect(Collectors.toList());
		}
		return Collections.emptyList();
	}

	@Override
	public void retry(ReliableMsgLog reliableMsg, int maxTimes) {
		reliableMsg.setRetryTimes(Ints.addI(reliableMsg.getRetryTimes(), 1));
		if (reliableMsg.getRetryTimes() > maxTimes) {
			this.executeMultiSql((conn) -> {
				String sql = "INSERT INTO " + failTableName
						+ "(MSG_ID,PHASE,RETRIED_TIMES,MAX_RTS,VERSION,CREATE_TIME,UPDATE_TIME,MESSAGE,QUEUE,QUEUE_TYPE)"
						+ " VALUES(?,?,?,?,?,?,?,?,?,?,?,?)";
				try {
					final byte[] serialize = App.serialize(reliableMsg.getMessage());
					executeUpdate(sql, reliableMsg.getMsgId(), reliableMsg.getPhase(), reliableMsg.getRetryTimes(),
							reliableMsg.getVersion(), reliableMsg.getCreateTime(), reliableMsg.getUpdateTime(),
							serialize, reliableMsg.getQueue(), reliableMsg.getQueueType());
				} catch (Exception e) {
					throw new ConsistencyException(e);
				}

				sql = "DELETE FROM " + tableName + " WHERE MSG_ID = ?";
				executeMultiSql_Update(conn, sql, reliableMsg.getMsgId());
			});
		} else {
			String sql = "UPDATE " + tableName + " SET RETRIED_TIMES = ? , UPDATE_TIME = ? WHERE MSG_ID = ?";
			executeUpdate(sql, reliableMsg.getRetryTimes(), new Date(), reliableMsg.getMsgId());
		}
	}
}
